package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {

    /**
     * 使用KStream处理消息
     *
     * @param streamsBuilder
     * @return
     */
    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        // 接收消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        stream.map((key, value) -> {
                    // 解析value
                    UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                    // 重新设置key和value的值
                    return new KeyValue<>(mess.getArticleId().toString(), mess.getType().name() + ":" + mess.getAdd());
                })
                .groupBy((key, value) -> value)     // 根据value统计结果
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))     // 时间窗口,每十秒汇报结果
                .aggregate(new Initializer<String>() {
                    /**
                     *初始化参数,创建的时候执行
                     * 完成统计数据的初始化以及上一次统计数据的清零
                     * @return 初始化数据
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    /**
                     *统计每个消息中的数据,每次有消息都会执行
                     * @param key       KStream中的key 文章的id
                     * @param value     KStream中的value  like:1
                     * @param aggValue  如果是第一个消息那么aggregate就是初始化的值  COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                     *                  如果不是第一个消息那么aggregate就是上一次统计的结果 COLLECTION:1,COMMENT:0,LIKES:0,VIEWS:0
                     * @return 统计后的结果
                     */
                    @Override
                    public String apply(String key, String value, String aggValue) {
                        // 判断value是否为空,如果为空,则返回上一次的结果
                        if (StringUtils.isEmpty(value)) {
                            return aggValue;
                        }

                        // 将统计结果赋值给变量存储
                        int col = 0, com = 0, lik = 0, vie = 0;
                        // aggValue
                        String[] splitAggValue = aggValue.split(","); // [COLLECTION:0,COMMENT:0]
                        for (String val : splitAggValue) {
                            String[] values = val.split(":"); // [COLLECTION, 0]
                            // 将结果赋值给变量
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(values[0])) {
                                case COLLECTION:
                                    col = Integer.parseInt(values[1]);
                                    break;
                                case COMMENT:
                                    com = Integer.parseInt(values[1]);
                                    break;
                                case LIKES:
                                    lik = Integer.parseInt(values[1]);
                                    break;
                                case VIEWS:
                                    vie = Integer.parseInt(values[1]);
                                    break;
                            }
                        }

                        // 根据value计算结果
                        String[] msgValue = value.split(":"); // [like:1]
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(msgValue[0])) {
                            case COLLECTION:
                                col += Integer.parseInt(msgValue[1]);
                                break;
                            case COMMENT:
                                com += Integer.parseInt(msgValue[1]);
                                break;
                            case LIKES:
                                lik += Integer.parseInt(msgValue[1]);
                                break;
                            case VIEWS:
                                vie += Integer.parseInt(msgValue[1]);
                                break;
                        }

                        // 数据格式转换,将计算后的结果添加到 COLLECTION:1,COMMENT:0,LIKES:0,VIEWS:0 中
                        String format = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                        System.out.println("文章的id:" + key);
                        System.out.println("当前时间窗口内的消息处理结果：" + format);
                        return format;
                    }
                }, Materialized.as("hot-atricle-stream-count-001"))
                .toStream()
                .map((key, value) -> {
                    System.out.println("key : " + key + " value: " + value);
                    return new KeyValue<>(key.key(), formatObj(key.key(), value));
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC); // 发送消息

        return stream;
    }

    /**
     * 将结果转换为json再发送到kafka
     *
     * @param key
     * @param value
     * @return
     */
    private String formatObj(String key, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.parseLong(key));

        String[] val = value.split(","); // [COMMENT, 1]
        switch (UpdateArticleMess.UpdateArticleType.valueOf(val[0])) {
            case COLLECTION:
                mess.setCollect(Integer.parseInt(val[1]));
                break;
            case COMMENT:
                mess.setComment(Integer.parseInt(val[1]));
                break;
            case LIKES:
                mess.setLike(Integer.parseInt(val[1]));
                break;
            case VIEWS:
                mess.setView(Integer.parseInt(val[1]));
                break;
        }

        return JSON.toJSONString(mess);
    }
}
